package org.example.message_queue.mqserver;

import org.example.message_queue.common.Consumer;
import org.example.message_queue.common.MqException;
import org.example.message_queue.mqserver.core.*;
import org.example.message_queue.mqserver.datacenter.DiskDataCenter;
import org.example.message_queue.mqserver.datacenter.MemoryDataCenter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/*
*   这个类表示虚拟主机
*   管理着自己的交换机、队列、绑定、消息
*   同时提供API给上层使用
*/
public class VirtualHost {
    private String virtualHostName;
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);

//    创建一个锁对象，加锁就对这个锁对象加锁
    private final Object exchangeLocker = new Object();
    private final Object queueLocker = new Object();

    public VirtualHost(String virtualHostName) {
        this.virtualHostName = virtualHostName;

//        memoryDataCenter一创建就会new，就已经初始化了，所以这里不需要初识化memoryDataCenter
        diskDataCenter.init();
//        异常向上抛但是总要有一层处理异常
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[virtualHost]恢复内存失败！");
        }
    }

    public String getVirtualHostName() {
        return virtualHostName;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    //    创建交换机
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete, Map<String,Object> arguments){
//        确定虚拟主机和交换机的重属关系，有三种方案，选择了交换机名称=虚拟机名+交换机名
        exchangeName = virtualHostName+exchangeName;
        try{
            synchronized (exchangeLocker){
//        1.判断交换机是否存在,直接通过内存查询
                Exchange existExchange = memoryDataCenter.getExchange(exchangeName);
//        已经存在
                if (existExchange != null){
                    System.out.println("[vVirtualHost]交换机已经存在！exchangeName="+exchangeName);
                    return true;
                }
//        不存在，就要创建交换机，将其添加到硬盘，内存中
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
//        假如durable为true，就是实现可持久化，就是写入硬盘；加入为false，就是不实现可持久化，那就不用写入硬盘
//        先写硬盘再写内存，是因为硬盘更容易写失败. 如果硬盘写失败了, 内存就不写了
//        要是先写内存, 内存写成功了, 硬盘写失败了, 还需要把内存的数据给再删掉. 就比较麻烦了.
                if (durable){
                    diskDataCenter.insertExchange(exchange);
                }
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost]交换机创建成功！ exchangeName="+exchangeName);

            }
            return true;
        } catch (Exception e){
            System.out.println("[VirtualHost]交换机创建失败！ exchangeName="+exchangeName);
            e.printStackTrace();
            return false;
        }
    }

//    删除交换机
    public boolean exchangeDelete(String exchangeName){
        exchangeName = virtualHostName+exchangeName;
        try{
            synchronized (exchangeLocker){
                //        1.查找交换机是否存在
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if (toDelete == null){
//                交换机不存在则无法删除
                    throw new MqException("[VirtualHost] 交换机不存在无法删除!");
                }
//            存在，就要从硬盘上删除，再删除内存中的数据
                if (toDelete.isDurable()){
                    diskDataCenter.deleteExchange(exchangeName);
                }
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost]交换机删除成功！ exchangeName="+exchangeName);
            }
            return true;
        } catch (MqException e) {
            System.out.println("[VirtualHost]交换机删除失败！ exchangeName="+exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueDeclare(String queueName, boolean durable,boolean autoDelete, boolean exclusive,Map<String,Object> arguments){
        queueName = virtualHostName + queueName;
        try{
            synchronized (queueLocker){
                //            1.判断队列是否存在,从内存中获取
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
//            已经存在的情况
                if (existsQueue != null){
                    System.out.println("[VirtualHost]队列已经存在 queueName="+queueName);
                    return true;
                }
//            不存在的情况
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setAutoDelete(autoDelete);
                queue.setExclusive(exclusive);
                queue.setArguments(arguments);
                if (durable){
                    diskDataCenter.insertQueue(queue);
                }
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost]队列创建成功 queueName="+queueName);

            }
            return true;
        } catch (IOException e) {
            System.out.println("[VirtualHost]队列创建失败 queueName="+queueName);
            e.printStackTrace();
            return false;
        }
    }

//    删除队列
    public boolean queueDelete(String queueName){
        queueName = virtualHostName + queueName;
        try{
            synchronized (queueLocker){
                MSGQueue toDelete = memoryDataCenter.getQueue(queueName);
                if (toDelete == null){
                    throw new MqException("[VirtualHost]队列不存在无法删除！");
                }
                if (toDelete.isDurable()){
                    diskDataCenter.deleteQueue(queueName);
                }
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost]队列删除成功！ queueName="+queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 删除队列失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

//    创建绑定
    public boolean queueBind(String exchangeName,String queueName,String bindingKey){
        exchangeName = virtualHostName + exchangeName;
        queueName = virtualHostName + queueName;
        try{
            synchronized (exchangeLocker){
                synchronized (queueLocker){
//        1.判断binding是否存在
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existsBinding != null){
                        throw new MqException("[VirtualHost]绑定已经存在 exchangeName="+exchangeName+"，queueName="+queueName);
                    }

//        2.验证bindingKey是否合法
                    if (!router.checkBindingKey(bindingKey)){
                        throw new MqException("[VirtualHost] bindingKey 非法！ bindingKey="+bindingKey);
                    }

//        3.创建binding
                    Binding binding= new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);

//        4.分别判断交换机和队列是否存在
                    Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                    if (existsExchange == null){
                        throw new MqException("[VirtualHost] 交换机不存在！exchangeName="+exchangeName);
                    }
                    MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                    if (existsQueue == null){
                        throw new MqException("[VirtualHost] 队列不存在 queueName="+queueName);
                    }

//        5.exchange和queue同时实现durable才能插入硬盘中
                    if(existsExchange.isDurable() && existsQueue.isDurable()){
                        diskDataCenter.insertBinding(binding);
                    }

                    memoryDataCenter.insertBinding(binding);
                }
            }
            System.out.println("[VirtualHost] 绑定创建成功 queueName="+queueName+",exchangeName="+exchangeName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost]绑定失败 exchangeName="+exchangeName+",queueName="+queueName);
            e.printStackTrace();
            return false;
        }
    }

//    删除绑定
    public boolean queueUnbind(String exchangeName,String queueName){
        exchangeName = virtualHostName + exchangeName;
        queueName = virtualHostName + queueName;

        try{
            Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
            if (existsBinding == null){
                throw new MqException("[VirtualHost] 绑定不存在，无法删除 exchangeName="+exchangeName+",queueName="+queueName);
            }
//            MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
//            if (existsQueue == null){
//                throw new MqException("[VirtualHost] 队列不存在， queueName="+queueName);
//            }
//            Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
//            if (existsExchange == null){
//                throw new MqException("[VirtualHost] 交换机不存在， exchangeName="+exchangeName);
//            }
//            if (existsExchange.isDurable() && existsQueue.isDurable()){
//                diskDataCenter.deleteBinding(existsBinding);
//            }
//            无论绑定是否持久化，都尝试从硬盘上删除绑定，就算不存在这个删除也无关紧要
            diskDataCenter.deleteBinding(existsBinding);
            memoryDataCenter.deleteBinding(existsBinding);
            System.out.println("[VirtualHost] 绑定删除成功！");
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost]删除绑定失败！");
            e.printStackTrace();
            return false;
        }
    }

//    public boolean basicpublish2(String exchangeName,String routingKey,BasicProperties basicProperties,byte[] body){
//        try{
//            exchangeName = virtualHostName + exchangeName;
//            if (router.checkRoutingKey(routingKey)){
//                throw new MqException("routingkey非法！");
//            }
//            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
//            if (exchange == null){
//                throw new MqException("交换机不存在！");
//            }
//            if (exchange.getType() == ExchangeType.DIRECT){
//                String queueName = virtualHostName + routingKey;
//                MSGQueue queue = memoryDataCenter.getQueue(queueName);
//                if (queue == null){
//                    throw new MqException("队列不存在");
//                }
////                队列存在,就可以存消息了，那就创建消息对象
//                Message message = Message.createMessageByWithId(routingKey,basicProperties,body);
////                if (message.getDeliverMode() == 2){
////                    diskDataCenter.sendMessage(queue,message);
////                }
////                memoryDataCenter.sendMessage(queue,message);
////                直接调用sendMessage方法就OK了，上述思想在方法中实现了
//                sendMessage(queue,message);
//            }else {
////              找到这个交换机关联的所有绑定，并遍历这些绑定对象
//                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
//                for (:Map.Entry)
//
//
//            }
//            return true;
//        } catch (Exception e) {
//            System.out.println("发送消息失败");
//            e.printStackTrace();
//            return false;
//        }
//    }

//    发送消息到指定的交换机/队列
    public boolean basicPublish(String exchangeName,String routingKey,BasicProperties basicProperties, byte[] body){
        try{
//            1.exchangeName进行转换
            exchangeName = virtualHostName + exchangeName;
//            2.检查routingKey是否合法
            if (!router.checkRoutingKey(routingKey)){
                throw new MqException("[virtualHost] routingKey非法！");
            }
//            3.根据交换机名字，找到交换机对象，判断是否存在，再根据交换机类型决定后续转发方式
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null){
                throw new MqException("[virtualHost] 交换机不存在 exchangeName="+exchangeName);
            }
//            交换机类型
//            （1）直接交换机，直接将routingKey作为队列的名字，将消息写入指定的队列
//            （2）fanout/topic交换机
            if (exchange.getType() == ExchangeType.DIRECT){
                String queueName = virtualHostName + routingKey;
                Message message = Message.createMessageByWithId(routingKey,basicProperties,body);
//                还要确保队列存在
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null){
                    throw new MqException("[virtualHost] 队列不存在 queueName="+queueName);
                }
//                队列存在，就写入消息
                sendMessage(queue,message);
            }else {
//              找到这个交换机关联的所有绑定，并遍历这些绑定对象
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                for (Map.Entry<String,Binding> entry : bindingsMap.entrySet()){
//                    (1)获取绑定对象，判定对应的队列是否存在
                    Binding binding = entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if (queue == null){
                        System.out.println("[virtualHost]basic发送消息时，发现队列不存在！queueName="+binding.getQueueName());
                        continue;
                    }
//                    (2)构造消息对象
                    Message message = Message.createMessageByWithId(routingKey,basicProperties,body);
//                    （3）判定这个消息是否能转发给该队列
                    if (!router.route(exchange.getType(),binding,message)){
                        continue;
                    }
//                    (4)真正转发消息给队列
                    sendMessage(queue,message);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("发送消息失败");
            e.printStackTrace();
            return false;
        }
    }

    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
//        往硬盘和内存中写入消息
        int deliverMode = message.getDeliverMode();
//        deliverMode : 1是不持久化  2是持久化
        if (deliverMode == 2){
            diskDataCenter.sendMessage(queue,message);
        }
        memoryDataCenter.sendMessage(queue,message);
//        通知队列有新消息，消费者可以消息了
        consumerManager.notifyConsumer(queue.getName());
    }

//    订阅消息：消费者订阅一个队列，假如这个队列有信息，就会往消费者推送
//    consumerTag：就是消费者的身份标识
//    queueName：是消费者订阅的这个队列
//    autoAck：信息被消费完成后的应答方式，true是自动应答，false是手动应答
//    consumer：是一个回调函数
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer){
//        构造ConsumerEnv对象，将对应的队列找到，再将Consumer对象添加到该队列中
        queueName = virtualHostName + queueName;
        try{
            consumerManager.addConsumer(consumerTag,queueName,autoAck,consumer);
            System.out.println("[virtualHost] basicConsum成功！ queueName="+queueName);
            return true;
        }catch (Exception e){
            System.out.println("[virtualHost] basicConsum失败！ queueName="+queueName);
            e.printStackTrace();
            return false;
        }

    }

    public boolean basicAck(String queueName,String messageId){
        queueName = virtualHostName + queueName;
        try{
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null){
                throw new MqException("[virtualHost] 要确认的消息不存在 messageId="+messageId);
            }
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null){
                throw new MqException("[virtualHost] 要确认的队列不存在 queueName="+queueName);
            }
            if (message.getDeliverMode() == 2){
                diskDataCenter.deleteMessage(queue,message);
            }
            memoryDataCenter.removeMessage(messageId);
            memoryDataCenter.removeMessageWaitAck(queueName,messageId);
            System.out.println("[virtualHost] basicAck 消息确认成功 queueName="+queueName+",messageId="+messageId);
            return true;
        }catch (Exception e){
            System.out.println("[virtualHost] basicAck 消息确认失败 queueName="+queueName+",messageId="+messageId);
            e.printStackTrace();
            return false;
        }
    }
}
